package com.whut.monitor.dao.impl;

import com.whut.monitor.constant.HbaseCell;
import com.whut.monitor.constant.HbaseRow;
import com.whut.monitor.constant.HbaseRows;
import com.whut.monitor.dao.IHBaseRestDao;
import com.whut.monitor.netty.handler.FileServerHandler;
import com.whut.monitor.util.Base64Util;
import com.whut.monitor.util.HttpRequestUtil;
import javafx.util.Pair;
import org.apache.log4j.Logger;
import org.codehaus.jettison.json.JSONArray;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * Created by YY on 2017-12-29.
 */
@Component
public class HBaseRestDaoImpl implements IHBaseRestDao {

    private static String uriBase = "http://59.69.101.206:6077/";

    private static final Logger logger = Logger.getLogger(FileServerHandler.class);

    /**
     * 获取所有表名
     *
     * @return 表名
     */
    @Override
    public JSONArray getHTableArray() {
        Pair<Integer, String> pair = HttpRequestUtil.httpGet(uriBase);
        JSONArray jsonArray = null;
        if (pair.getKey() == 200) {
            try {
                jsonArray = new JSONObject(pair.getValue()).getJSONArray("table");
            } catch (JSONException e) {
                e.printStackTrace();
            }
        }
        return jsonArray;
    }

    /**
     * 列出表的域(regions)
     *
     * @param tableName 输入表名
     * @return 结构
     */
    @Override
    public JSONObject getHTableRegions(String tableName) {
        Pair<Integer, String> pair = HttpRequestUtil.httpGet(uriBase.concat(tableName).concat("/regions"));
        JSONObject jsonObject = null;
        if (pair.getKey() == 200) {
            try {
                jsonObject = new JSONObject(pair.getValue());
            } catch (JSONException e) {
                e.printStackTrace();
            }
        }
        return jsonObject;
    }

    /**
     * 获取某个表的结构
     *
     * @param tableName 输入表名
     * @return 结构
     */
    @Override
    public JSONObject getHTableSchema(String tableName) {
        Pair<Integer, String> pair = HttpRequestUtil.httpGet(uriBase.concat(tableName).concat("/schema"));
        JSONObject object = null;
        if (pair.getKey() == 200) {
            try {
                object = new JSONObject(pair.getValue());
            } catch (JSONException e) {
                e.printStackTrace();
            }
        }
        return object;
    }

    /**
     * 通过JSON格式创建一张表
     *
     * @param tableName 输入表名
     * @param strJson   JSON对象
     * @return 创建结果
     */
    @Override
    public Pair<Boolean, String> createHTableByJson(String tableName, JSONObject strJson) {
        String strUrl = uriBase.concat(tableName).concat("/schema");
        Pair<Integer, String> pair = HttpRequestUtil.httpPutByJson(strUrl, strJson);
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            return new Pair(true, pair.getValue());
        }
        return new Pair(false, pair.getValue());
    }

    /**
     * 通过JSON格式更新表结构
     *
     * @param tableName 输入表名
     * @param strJson   JSON对象
     * @return 更新结果
     */
    @Override
    public Pair<Boolean, String> updateHTableByJson(String tableName, JSONObject strJson) {
        String strUrl = uriBase.concat(tableName).concat("/schema");
        Pair<Integer, String> pair = HttpRequestUtil.httpPostByJson(strUrl, strJson);
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            return new Pair(true, pair.getValue());
        }
        return new Pair(false, pair.getValue());
    }

    /**
     * 通过表名删除一张表
     *
     * @param tableName 输入表名
     * @return 删除结果
     */
    @Override
    public Pair<Boolean, String> deleteHTable(String tableName) {
        String strUrl = uriBase.concat(tableName).concat("/schema");
        Pair<Integer, String> pair = HttpRequestUtil.httpDelete(strUrl);
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            return new Pair(true, pair.getValue());
        }
        return new Pair(false, pair.getValue());
    }

    /**
     * 获得一行
     *
     * @param tableName 输入表名
     * @param rowkey    行键
     * @return 指定行的Json对象
     */
    @Override
    public Pair<Boolean, String> findRow(String tableName, String rowkey) {

        String strUrl = uriBase.concat(tableName).concat("/").concat(rowkey);
        Pair<Integer, String> pair = HttpRequestUtil.httpGet(strUrl);
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            return new Pair(true, pair.getValue());
        }
        return new Pair(false, pair.getValue());
    }

    /**
     * 获得多行
     *
     * @param tableName    输入表名
     * @param prefixFilter 行键的前缀
     * @param maxCount     最大的cell数量，注意不是行的数量
     * @return 查找到的Json对象
     */
    @Override
    public Pair<Boolean, String> findRowsByprefixFilter(String tableName, String prefixFilter, int maxCount) {
        String strUrl = uriBase.concat(tableName).concat("/").concat("scanner/");
        String scannerUrl = "";
        String xmlStr = "<Scanner batch=" + '"' + maxCount + '"' + ">";
        xmlStr += "<filter>{";
        xmlStr += ('"' + "type" + '"' + ":" + '"' + "PrefixFilter" + '"' + ',');
        xmlStr += ('"' + "value" + '"' + ":" + '"' + prefixFilter + '"' + '}');
        xmlStr += "</filter></Scanner>";
        System.out.println(xmlStr);
        Pair<Integer, String> pair = HttpRequestUtil.httpPutReHeaderByXml(strUrl, xmlStr, "Location");
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            scannerUrl = pair.getValue();
            pair = HttpRequestUtil.httpGet(scannerUrl);
        }
        System.out.println("释放scanner:" + HttpRequestUtil.httpDelete(scannerUrl));
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            return new Pair(true, pair.getValue());
        }
        return new Pair(false, pair.getValue());
    }

    /**
     * 获得多行
     *
     * @param tableName    输入表名
     * @param suffixFilter 行键的后缀
     * @param maxCount     最大的cell数量，注意不是行的数量
     * @return 查找到的Json对象
     */
    @Override
    public Pair<Boolean, String> findRowsBySuffixFilter(String tableName, String suffixFilter, int maxCount) {
        String strUrl = uriBase.concat(tableName).concat("/").concat("scanner/");
        String scannerUrl = "";
        String xmlStr = "<Scanner batch=" + '"' + maxCount + '"' + ">";
        xmlStr += "<filter>{";
        xmlStr += ('"' + "type" + '"' + ":" + '"' + "SuffixFilter" + '"' + ',');
        xmlStr += ('"' + "value" + '"' + ":" + '"' + suffixFilter + '"' + '}');
        xmlStr += "</filter></Scanner>";
        System.out.println(xmlStr);
        Pair<Integer, String> pair = HttpRequestUtil.httpPutReHeaderByXml(strUrl, xmlStr, "Location");
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            scannerUrl = pair.getValue();
            pair = HttpRequestUtil.httpGet(scannerUrl);
        }
        System.out.println("释放scanner:" + HttpRequestUtil.httpDelete(scannerUrl));
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            return new Pair(true, pair.getValue());
        }
        return new Pair(false, pair.getValue());
    }

    /**
     * 获得一列
     *
     * @param tableName    输入表名
     * @param rowkey       行键名
     * @param columnFamily 列族名
     * @param column       列名
     * @return 指定列的Json对象
     */
    @Override
    public Pair<Boolean, String> findCell(String tableName, String rowkey, String columnFamily, String column) {

        String strUrl = uriBase.concat(tableName).concat("/").concat(rowkey).concat("/").
                concat(columnFamily).concat(":").concat(column);
        Pair<Integer, String> pair = HttpRequestUtil.httpGet(strUrl);
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            return new Pair(true, pair.getValue());
        }
        return new Pair(false, pair.getValue());
    }

    /**
     * 获得一列的前n个版本
     *
     * @param tableName    输入表名
     * @param rowkey       行键名
     * @param columnFamily 列族名
     * @param column       列名
     * @param n            前n个版本
     * @return 指定列的Json对象
     */
    @Override
    public Pair<Boolean, String> findMultilVersionCell(String tableName, String rowkey, String columnFamily, String column, int n) {

        String strUrl = uriBase.concat(tableName).concat("/").concat(rowkey).concat("/").
                concat(columnFamily).concat(":").concat(column).concat("?v=") + n;
        Pair<Integer, String> pair = HttpRequestUtil.httpGet(strUrl);
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            return new Pair(true, pair.getValue());
        }
        return new Pair(false, pair.getValue());
    }

    /**
     * 插入一个单元cell
     *
     * @param tableName    输入表名
     * @param rowkey       行键
     * @param columnfamily 列族
     * @param column       列
     * @param value        值
     * @return 插入结果
     */
    @Override
    public Pair<Boolean, String> insertCell(String tableName, String rowkey, String columnfamily, String column, String value) {
        String strUrl = uriBase.concat(tableName).concat("/").concat(rowkey);
        HbaseCell hbaseCell = new HbaseCell(columnfamily, column, value);
        HbaseRow hbaseRow = new HbaseRow(rowkey, hbaseCell);
        HbaseRows hbaseRows = new HbaseRows(hbaseRow);
        return insertRow("test", rowkey, hbaseRows.getHbaseRowsJson());
    }

    /**
     * 插入一行
     *
     * @param tableName  输入表名
     * @param rowkey     行键
     * @param jsonObject 插入的值
     * @return 插入结果
     */
    @Override
    public Pair<Boolean, String> insertRow(String tableName, String rowkey, JSONObject jsonObject) {
        String strUrl = uriBase.concat(tableName).concat("/").concat(rowkey);
        Pair<Integer, String> pair = HttpRequestUtil.httpPutByJson(strUrl, jsonObject);
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            return new Pair(true, pair.getValue());
        }
        return new Pair(false, pair.getValue());
    }

    /**
     * 插入多行
     *
     * @param tableName  输入表名
     * @param jsonObject 插入的值
     * @return 插入结果
     */
    @Override
    public Pair<Boolean, String> insertRows(String tableName, JSONObject jsonObject) {
        String strUrl = uriBase.concat(tableName).concat("/").concat("rows");
        Pair<Integer, String> pair = HttpRequestUtil.httpPutByJson(strUrl, jsonObject);
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            return new Pair(true, pair.getValue());
        }
        return new Pair(false, pair.getValue());
    }

    //兼容原版的Hbase操作
    //插入单个值
    @Override
    public Boolean add(String tableName, String rowKey, String family, String qualifier, String value) {
        logger.info("add:" + tableName + rowKey + family + qualifier + value);
        return insertCell(tableName, rowKey, family, qualifier, value).getKey();
    }

    @Override
    public String find(String tableName, String rowKey, String family, String qualifier) {
        logger.info("find:" + tableName + rowKey + family + qualifier);
        return findCell(tableName, rowKey, family, qualifier).getValue();
    }

    //插入一行，多列
    @Override
    public Boolean adds(String tableName, String rowKey, String family, Map<String, String> quaAndValue) {

        logger.info("adds:" + tableName + rowKey + family);
        HbaseRow hbaseRow = new HbaseRow(rowKey);
        for (String column : quaAndValue.keySet()) {
            hbaseRow.addHbaseCell(new HbaseCell(family, column, quaAndValue.get(column)));
        }
        HbaseRows hbaseRows = new HbaseRows(hbaseRow);
        return insertRow(tableName, rowKey, hbaseRows.getHbaseRowsJson()).getKey();
    }

    //插入多行一列
    @Override
    public Boolean adds(String tableName, Map<String, String> rowkeyAndValue, String family, String qualifier) {
        logger.info("adds:" + tableName + family + qualifier);
        HbaseRows hbaseRows = new HbaseRows();
        for (String rowkey : rowkeyAndValue.keySet()) {
            hbaseRows.addHbaseRow(new HbaseRow(rowkey, new HbaseCell(family, qualifier, rowkeyAndValue.get(rowkey))));
        }
        return insertRows(tableName, hbaseRows.getHbaseRowsJson()).getKey();
    }

    @Override
    public Boolean finds(String tableName, String startRow, String stopRow, Map<String, String> outputRowkeyAndValue) {

        logger.info("finds:" + tableName + startRow + stopRow);
        String strUrl = uriBase.concat(tableName).concat("/").concat("scanner/");
        String scannerUrl = "";
        String xmlStr = "<Scanner";
        xmlStr += (" startRow=" + '"' + Base64Util.encodeByBase64(startRow) + '"');
        xmlStr += (" endRow=" + '"' + Base64Util.encodeByBase64(stopRow) + '"');
        xmlStr += "></Scanner>";
        System.out.println(xmlStr);
        Pair<Integer, String> pair = HttpRequestUtil.httpPutReHeaderByXml(strUrl, xmlStr, "Location");
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            scannerUrl = pair.getValue();
            pair = HttpRequestUtil.httpGet(scannerUrl);
            HttpRequestUtil.httpDelete(scannerUrl);
        }
        if (pair.getKey() == 200 || pair.getKey() == 201) {
            List<HbaseRow> hbaseRowList = HbaseRows.parseHBaseRowsJson(pair.getValue());
            for (HbaseRow hbaseRow : hbaseRowList) {
                List<HbaseCell> hbaseCellList = hbaseRow.getHbaseCellList();
                if (!hbaseCellList.isEmpty()) {
                    outputRowkeyAndValue.put(hbaseRow.getRowKey(), hbaseCellList.get(0).value);
                }
            }
            return true;
        }
        return false;
    }
}


